﻿using Grow.Binary;
using Grow.Domain.DomainRepository;
using Grow.Domain.DomainRepository.CreateRepository;
using Grow.Domain.EventBus;
using Grow.Domain.EventBus.CreateEventBus;
using Grow.EventBus.Cap.DomainEventHandlerInvoker;
using Grow.EventBus.Cap.DomainEventSend;
using Grow.EventBus.Cap.Jobs;
using Grow.EventBus.Cap.Kafka.CreateEventBus;
using Grow.EventBus.Cap.PublisherDomainEvents;
using Grow.EventBus.Cap.SubscriberDomainEvents;
using Grow.Json;
using Grow.Module;
using Grow.Module.Module;
using Grow.Module.Service;
using Grow.Scheduler;
using Grow.Scheduler.CreateScheduler;
using Microsoft.Extensions.DependencyInjection;

namespace Grow.EventBus.Cap.Kafka
{
    public abstract  class EventBusKafKaModuleBase : ModuleBase, IDomainEventBusModule
    {
        public override ModuleLevel Level =>  ModuleLevel.Business;

        public override int Order => 5;

        public override void AddDependModule(IDependModuleBuilder dependModuleBuilder)
        {
            dependModuleBuilder.AddDependModule<IRepositoryModule>();
            dependModuleBuilder.AddDependModule<IBinaryModule>();
            dependModuleBuilder.AddDependModule<IJsonModule>();
            dependModuleBuilder.AddDependModule<ITaskSchedulerModule>();
        }

      

        public override void AddService(IServiceContainer serviceContainer)
        {
            var createEventBusBuilder = new CreateEventBusBuilder(serviceContainer);
            CreateEventBus(createEventBusBuilder);
            serviceContainer.BuilderManager.SetBuilder<ICreateEventBusBuilder>(createEventBusBuilder);

            var createRepositoryBuilder = serviceContainer.BuilderManager.GetRequiredBuilder<ICreateRepositoryBuilder>();
            createRepositoryBuilder.AddDbMap<PublisherDomainEventDbMap>();
            createRepositoryBuilder.AddDbMap<SubscriberDomainEventDbMap>();
            createRepositoryBuilder.AddRepository<IPublisherDomainEventRepository, PublisherDomainEventRepository, PublisherDomainEvent>();
            createRepositoryBuilder.AddRepository<ISubscriberDomainEventRepository, SubscriberDomainEventRepository, SubscriberDomainEvent>();


            var createSchedulerBuilder = serviceContainer.BuilderManager.GetRequiredBuilder<ICreateSchedulerBuilder>();
            createSchedulerBuilder.AddJob<ExpirerPublisherDomainEventJob>();
            createSchedulerBuilder.AddJob<ExpirerSubscriberDomainEventJob>();
            createSchedulerBuilder.AddJob<RetryPublisherDomainEventJob>();
            createSchedulerBuilder.AddJob<RetrySubscriberDomainEventJob>();


            serviceContainer.ServiceCollection.AddScoped<IDomainEventPublisher, DomainEventPublisher>();
            serviceContainer.ServiceCollection.AddScoped<IPublisherDomainEventHandler, PublisherDomainEventHandler>();
            serviceContainer.ServiceCollection.AddScoped<IPublisherDomainEventService, PublisherDomainEventService>();


            serviceContainer.ServiceCollection.AddScoped<IDomainEventSubscriber, DomainEventSubscriber>();
            serviceContainer.ServiceCollection.AddScoped<ISubscriberDomainEventHandler, SubscriberDomainEventHandler>();
            serviceContainer.ServiceCollection.AddScoped<ISubscriberDomainEventService, SubscriberDomainEventService>();

            serviceContainer.ServiceCollection.AddScoped<IDomainEventSend, DomainEventSend>();
            serviceContainer.ServiceCollection.AddScoped<IDomainEventHandlerInvoker, DomainEventHandlerInvoker.DomainEventHandlerInvoker>();

            serviceContainer.ServiceCollection.AddSingleton<IKafkaConnection, KafkaConnection>();
        }

      

        public override void UseService(IServiceContext serviceContext)
        {
            var eventBusOptions = serviceContext.GetOptions<DomainEventBusOptions>();
            var cancellationTokenContext = serviceContext.GetCancellationTokenContext();
            var scheduler = serviceContext.GetRequiredService<IScheduler>();

            scheduler.AddJob<RetryPublisherDomainEventJob>($"{eventBusOptions.EnvironmentName}_RetryPublisherEventData", eventBusOptions.RetryCron, cancellationTokenContext.Token);
            scheduler.AddJob<RetrySubscriberDomainEventJob>($"{eventBusOptions.EnvironmentName}_RetrySubscriberEventData", eventBusOptions.RetryCron, cancellationTokenContext.Token);
            scheduler.AddJob<ExpirerPublisherDomainEventJob>($"{eventBusOptions.EnvironmentName}_ExpirerPublisherEventDataJob", eventBusOptions.ExpirerCron, cancellationTokenContext.Token);
            scheduler.AddJob<ExpirerSubscriberDomainEventJob>($"{eventBusOptions.EnvironmentName}_ExpirerPublisherEventDataJob", eventBusOptions.ExpirerCron, cancellationTokenContext.Token);

            var subscriberDomainEvents = serviceContext.GetServices<ISubscriberDomainEvent>();
            Parallel.ForEach(subscriberDomainEvents, subscriberDomainEvent =>
            {
                _ = Task.Factory.StartNew(async () =>
                {
                    await subscriberDomainEvent.SubscriberAsync(cancellationTokenContext.Token);
                }, cancellationTokenContext.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
            });
        }

        public abstract void CreateEventBus(ICreateEventBusBuilder createEventBusBuilder);
    }
}
